package mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class FlowCountSortDriver extends Configured implements Tool {
    /**
     * 将自己的 map  Reduce 代码添加到框架中
     */
    @Override
    public int run(String[] strings) throws Exception {

        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        //1.实例一个 Job
        Job job = Job.getInstance(conf, "FlowCount");
        //2.设置读取数据的class
        job.setInputFormatClass(TextInputFormat.class);
        //  设置读取数据的路径
        TextInputFormat.addInputPath(job,new Path("/laoshuming/week_01/input/HTTP_20130313143750.dat"));

        //3.设置map
        job.setMapperClass(FlowCountMap.class);
        //  设置map输出的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(PhoneFlow.class);

        //4.设置reduce
        job.setReducerClass(FlowCountReduce.class);
        //  设置reduce的输出的类型
        job.setOutputKeyClass(PhoneFlow.class);
        job.setOutputValueClass(Text.class);

        //5.设置输出数据的class
        job.setOutputFormatClass(TextOutputFormat.class);
        //  设置输出数据的路径
        Path path = new Path("/laoshuming/week_01/temp");
        if(fs.exists(path))//如果目录存在，则删除目录
        {
            fs.delete(path,true);
        }
        TextOutputFormat.setOutputPath(job,path);
        //6.等待代码执行（返回状态码）

        if(job.waitForCompletion(true) ){
            //1.实例一个 Job
            Job jobSort = Job.getInstance(conf, "FlowSort");
            //2.设置读取数据的class
            jobSort.setInputFormatClass(TextInputFormat.class);
            //  设置读取数据的路径
            TextInputFormat.addInputPath(jobSort,new Path("/laoshuming/week_01/temp/part-r-00000"));
            //3.设置map
            jobSort.setMapperClass(FlowSortMap.class);
            //  设置map输出的类型
            jobSort.setMapOutputKeyClass(PhoneFlow.class);
            jobSort.setMapOutputValueClass(NullWritable.class);

            //4.设置reduce
            jobSort.setReducerClass(FlowSortReduce.class);
            //  设置reduce的输出的类型
            jobSort.setOutputKeyClass(PhoneFlow.class);
            jobSort.setOutputValueClass(NullWritable.class);

            //5.设置输出数据的class
            jobSort.setOutputFormatClass(TextOutputFormat.class);
            //  设置输出数据的路径
            Path sortPath = new Path("/laoshuming/week_01/output");
            if(fs.exists(sortPath))//如果目录存在，则删除目录
            {
                fs.delete(sortPath,true);
            }
            TextOutputFormat.setOutputPath(jobSort,sortPath);
            return jobSort.waitForCompletion(true)?0:1;
        }
        return 1;
    }

    public static void main(String[] args) throws Exception {
        //调用执行
        int run = ToolRunner.run(new FlowCountSortDriver(), args);
    }
}